#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <time.h>
#include <unistd.h>

// 消息结构体
typedef struct Message {
    int id;
    char content[256];
    struct Message* next;
} Message;

// 消息队列结构体
typedef struct {
    Message* head;
    Message* tail;
    int size;
    int capacity;
    int next_id; // 将ID计数器移到结构体中，并用互斥锁保护
    pthread_mutex_t mutex;
    pthread_cond_t not_empty;
    pthread_cond_t not_full;
} MessageQueue;

// 初始化消息队列
MessageQueue* mq_init(int capacity) {
    MessageQueue* mq = (MessageQueue*)malloc(sizeof(MessageQueue));
    if (!mq) {
        perror("Failed to allocate message queue");
        return NULL;
    }
    
    mq->head = NULL;
    mq->tail = NULL;
    mq->size = 0;
    mq->capacity = capacity;
    mq->next_id = 0; // 初始化ID计数器
    
    if (pthread_mutex_init(&mq->mutex, NULL) != 0) {
        perror("Mutex init failed");
        free(mq);
        return NULL;
    }
    
    if (pthread_cond_init(&mq->not_empty, NULL) != 0 ||
        pthread_cond_init(&mq->not_full, NULL) != 0) {
        perror("Condition init failed");
        pthread_mutex_destroy(&mq->mutex);
        free(mq);
        return NULL;
    }
    
    return mq;
}

// 销毁消息队列
void mq_destroy(MessageQueue* mq) {
    if (!mq) return;
    
    pthread_mutex_lock(&mq->mutex);
    
    Message* current = mq->head;
    while (current) {
        Message* temp = current;
        current = current->next;
        free(temp);
    }
    
    pthread_mutex_unlock(&mq->mutex);
    pthread_mutex_destroy(&mq->mutex);
    pthread_cond_destroy(&mq->not_empty);
    pthread_cond_destroy(&mq->not_full);
    
    free(mq);
}

// 添加消息到队列
int mq_enqueue(MessageQueue* mq, const char* content) {
    if (!mq || !content) return -1;
    
    pthread_mutex_lock(&mq->mutex);
    
    // 等待队列有空闲空间
    while (mq->size >= mq->capacity) {
        pthread_cond_wait(&mq->not_full, &mq->mutex);
    }
    
    // 创建新消息
    Message* new_msg = (Message*)malloc(sizeof(Message));
    if (!new_msg) {
        pthread_mutex_unlock(&mq->mutex);
        perror("Failed to allocate message");
        return -1;
    }
    
    // 使用受保护的ID计数器
    new_msg->id = mq->next_id++;
    snprintf(new_msg->content, sizeof(new_msg->content), "%s", content);
    new_msg->next = NULL;
    
    // 添加到队列尾部
    if (mq->tail) {
        mq->tail->next = new_msg;
        mq->tail = new_msg;
    } else {
        mq->head = mq->tail = new_msg;
    }
    
    mq->size++;
    
    printf("Produced: ID=%d, Content=%s, Queue Size=%d\n", 
           new_msg->id, new_msg->content, mq->size);
    
    // 通知消费者有新消息
    pthread_cond_signal(&mq->not_empty);
    pthread_mutex_unlock(&mq->mutex);
    
    return 0;
}

// 从队列获取消息
Message* mq_dequeue(MessageQueue* mq) {
    if (!mq) return NULL;
    
    pthread_mutex_lock(&mq->mutex);
    
    // 等待队列不为空
    while (mq->size == 0) {
        pthread_cond_wait(&mq->not_empty, &mq->mutex);
    }
    
    // 从队列头部取出消息
    Message* msg = mq->head;
    if (mq->head == mq->tail) {
        mq->head = mq->tail = NULL;
    } else {
        mq->head = mq->head->next;
    }
    
    mq->size--;
    
    printf("Consumed: ID=%d, Content=%s, Queue Size=%d\n", 
           msg->id, msg->content, mq->size);
    
    // 通知生产者有空闲空间
    pthread_cond_signal(&mq->not_full);
    pthread_mutex_unlock(&mq->mutex);
    
    return msg;
}

// 生产者线程函数
void* producer(void* arg) {
    MessageQueue* mq = (MessageQueue*)arg;
    char buffer[256];
    int count = 0;
    
    // 使用线程安全的随机数生成
    unsigned int seed = (unsigned int)(time(NULL) ^ pthread_self());
    
    while (count < 10000) {
        snprintf(buffer, sizeof(buffer), "Message %d from producer %lu", 
                 count++, (unsigned long)pthread_self());
        mq_enqueue(mq, buffer);
        usleep(rand_r(&seed) % 100000); // 使用线程安全的随机数
    }
    
    return NULL;
}

// 消费者线程函数
void* consumer(void* arg) {
    MessageQueue* mq = (MessageQueue*)arg;
    int count = 0;
    
    // 使用线程安全的随机数生成
    unsigned int seed = (unsigned int)(time(NULL) ^ pthread_self());
    
    while (1) {
        Message* msg = mq_dequeue(mq);
        if (msg) {
            free(msg);
            count++;
        }
        usleep(rand_r(&seed) % 150000); // 使用线程安全的随机数
    }
    
    return NULL;
}

int main() {
    // 创建消息队列，容量为5
    MessageQueue* mq = mq_init(100);
    if (!mq) {
        fprintf(stderr, "Failed to initialize message queue\n");
        return EXIT_FAILURE;
    }
    
    pthread_t prod_thread1, prod_thread2;
    pthread_t cons_thread1, cons_thread2;
    int ret = EXIT_SUCCESS;
    
    // 创建生产者线程
    if (pthread_create(&prod_thread1, NULL, producer, mq) != 0) {
        perror("Failed to create producer thread 1");
        ret = EXIT_FAILURE;
        goto cleanup;
    }
    
    if (pthread_create(&prod_thread2, NULL, producer, mq) != 0) {
        perror("Failed to create producer thread 2");
        ret = EXIT_FAILURE;
        pthread_cancel(prod_thread1);
        goto cleanup;
    }
    
    // 创建消费者线程
    if (pthread_create(&cons_thread1, NULL, consumer, mq) != 0) {
        perror("Failed to create consumer thread 1");
        ret = EXIT_FAILURE;
        pthread_cancel(prod_thread1);
        pthread_cancel(prod_thread2);
        goto cleanup;
    }
    
    if (pthread_create(&cons_thread2, NULL, consumer, mq) != 0) {
        perror("Failed to create consumer thread 2");
        ret = EXIT_FAILURE;
        pthread_cancel(prod_thread1);
        pthread_cancel(prod_thread2);
        pthread_cancel(cons_thread1);
        goto cleanup;
    }
    
    // 等待线程结束
    pthread_join(prod_thread1, NULL);
    pthread_join(prod_thread2, NULL);
    pthread_join(cons_thread1, NULL);
    pthread_join(cons_thread2, NULL);

cleanup:
    // 销毁消息队列
    mq_destroy(mq);
    
    if (ret == EXIT_SUCCESS) {
        printf("Message queue demo completed successfully.\n");
    } else {
        fprintf(stderr, "Message queue demo failed.\n");
    }
    
    return ret;
}
